package com.geekstudy.week9;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPubSub;


import java.util.EventObject;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class DistributedEventPublisher {

    private final SimplePublisher<EventObject> simplePublisher;

    // private final JedisPool jedisPool;
    // https://blog.csdn.net/ksisn/article/details/112797970
    private Watcher watcher;
    private ZooKeeper zooKeeper;

    private final ExecutorService executorService;

    public DistributedEventPublisher(String uri) {
        simplePublisher = new SimplePublisher();
        // this.jedisPool = new JedisPool(uri);
        try {
            this.watcher = new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    System.out.println("Watch state:" + watchedEvent.getState() + " type:" + watchedEvent.getType());
                }
            };
            this.zooKeeper = new ZooKeeper("127.0.0.1:2181", 4000, this.watcher);
            // zooKeeper.create("/week9Event");


        } catch(Exception e) {
            e.printStackTrace();
        }


        // Build-in listener
        addEventListener(event -> {
            if (event instanceof DistributedEventObject) {
                // Event -> Pub/Sub
                //Jedis jedis = jedisPool.getResource();
                //jedis.publish("test", (String) event.getSource());
                //jedis.close();
                String message = (String)event.getSource();
                try {
                    this.zooKeeper.create("/week9Event", message.getBytes("utf-8"), null, CreateMode.PERSISTENT);
                } catch(Exception e) {
                    e.printStackTrace();
                }

            }
        });

        this.executorService = Executors.newSingleThreadExecutor();

        /*
        executorService.execute(() -> {
            Jedis jedis = jedisPool.getResource();
            jedis.subscribe(new JedisPubSub() {
                @Override
                public void onMessage(String channel, String message) {
                    if ("test".equals(channel)) {
                        publish(new EventObject(message));
                    }
                }
            }, "test");
            jedis.close();
        });
        */
        executorService.execute(() -> {

        });
    }


    public void publish(Object event) {
//        simplePublisher.publish(new EventObject(event));
        simplePublisher.publish(new DistributedEventObject(event));
    }

    private void publish(EventObject event) {
//        simplePublisher.publish(new EventObject(event));
        simplePublisher.publish(event);
    }

    public void addEventListener(EventListenerApi eventListener) {
        simplePublisher.subscribe(new ListenerSubscriberAdapter(eventListener));
    }

    public void close() {
        // jedisPool.close();
        try {
            zooKeeper.close();
        } catch(Exception e) {
            e.printStackTrace();
        }
        executorService.shutdown();
    }







}
